/**
 * crawler
 *
 * outlook@China software studio
 * Copyright (c) 2008-2010. All Rights Reserved.
 * http://www.outlook.com
 */
package org.outlook.common.queue;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.outlook.common.log.LogUtil;

/**
 * function:
 * 
 * @author Watson email: watson.wu@hotmail.com
 * @date 2011
 */

@SuppressWarnings("unchecked")
public class ThreadPoolManager<T extends Task> {
	
	private Log logger = LogFactory.getLog(ThreadPoolManager.class);

	private static ThreadPoolManager tpm = new ThreadPoolManager();

	// 线程池维护线程的最少数量
	private int corePoolSize = 5;
	
	// 线程池维护线程的最大数量
	private int maxPoolSize = 20;
	
	// 线程池维护线程所允许的空闲时间
	private int keepAliveTime = 0;
	
	// 线程池所使用的缓冲队列大小
	private int workQueueSize = 10;

	// 消息缓冲队列
	Queue<Object> msgQueue = new LinkedList<Object>();

	// 访问消息缓存的调度线程
	final Runnable accessBufferThread = new Runnable() {
		public void run() {
			// 查看是否有待定请求，如果有，则创建一个新的Task，并添加到线程池中
			if (hasMoreAcquire()) {
				LogUtil.trace(logger, "============== ", getQueueSize(), " tasks need process ==============");
				Object msg = (Object) msgQueue.poll();
				Runnable task = (T) msg;
				threadPool.execute(task);
			}
			LogUtil.trace(logger, "============== no task need process, try sleep 10ns ==============");
			sleep(10);
		}

		private void sleep(int i) {
			try {
				Thread.sleep(i);
			} catch (InterruptedException e) {
				LogUtil.debug(logger, e.getMessage());
			}
		}
	};

	public int getQueueSize() {
		return msgQueue.size();
	}

	final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			LogUtil.debug(logger, "put ", (Task) r, " into queue and wait for excuting");
			msgQueue.offer((Task) r);
		}
	};

	// 管理线程池
	final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
			corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
			new ArrayBlockingQueue(workQueueSize), this.handler);

	// 调度线程池
	final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

	final ScheduledFuture taskHandler = scheduler.scheduleAtFixedRate(
			accessBufferThread, 0, 1, TimeUnit.SECONDS);

	public static ThreadPoolManager newInstance() {
		return tpm;
	}

	private ThreadPoolManager() {
	}

	private boolean hasMoreAcquire() {
		return !msgQueue.isEmpty();
	}

	public void add(T task) {
		//TODO
		threadPool.execute(task);
	}
	
	public boolean shutdown(boolean immediate) {
		if(immediate) {
			msgQueue.clear();
			threadPool.shutdownNow();
		} else {
			while(getQueueSize() == 0) {
			}
			threadPool.shutdown();
		}
		return false;
	}
}
